package data.platform.stream.internal.command;

import com.fasterxml.jackson.databind.ObjectMapper;
import data.platform.common.service.command.RulePointCommandService;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.common.stream.aggregates.*;
import data.platform.common.stream.valueobjects.RuleMetricCondition;
import data.platform.common.stream.valueobjects.WindowType;
import data.platform.common.util.DateUtil;
import data.platform.stream.internal.cache.RuleAggregateCache;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@AllArgsConstructor
public class RuleTask implements Runnable {

    private Rule rule;

    private MetricTagQueryService metricTagQueryService;

    private RuleAggregateCache ruleAggregateCache;

    private RulePointCommandService rulePointCommandService;

    private RuleScheduleServiceImpl ruleScheduleServiceImpl;

    private static final ObjectMapper objectMapper = new ObjectMapper();

    private static final ThreadLocal<SimpleDateFormat> dateFormatter = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));

    @Override
    public void run() {
        log.info("start rule task process, time is: {},  {}.", dateFormatter.get().format(new Date()), rule);

        try {
            List<RuleMetricCondition> ruleMetricConditions = rule.getRuleMetricConditions();
            long endTime = new Date().getTime();
            // 遍历任务的所有时间窗口，如果结束时间小于当前时间，执行标签计算，完成后，从缓存中删除值
            if (ruleAggregateCache.getCache().containsKey(rule.getId())) {
                Map<TimeWindow, Map<String, Map<Integer, WindowAggregateItem>>> timeWindowCache = ruleAggregateCache.getCache().get(rule.getId());
                Iterator<TimeWindow> timeWindowIterator = timeWindowCache.keySet().iterator();
                while (timeWindowIterator.hasNext()) {
                    TimeWindow timeWindow = timeWindowIterator.next();
                    if (timeWindow.getEnd() < endTime) {
                        Map<String, Map<Integer, WindowAggregateItem>> metricCache = timeWindowCache.get(timeWindow);
                        Map<Integer, WindowResult> comparisonResult = rule.apply(metricCache);

                        for (RuleMetricCondition ruleMetricCondition : ruleMetricConditions) {
                            String metric = ruleMetricCondition.getMetric();
                            if (metricCache.containsKey(metric)) {
                                for (Map.Entry<Integer, WindowAggregateItem> tagEntry : metricCache.get(metric).entrySet()) {
                                    // 从store删除数据
                                    Integer tagId = tagEntry.getKey();
                                    ruleAggregateCache.clearDB(rule.getId(), timeWindow, metric, tagId);
                                }
                            }
                        }

                        timeWindowIterator.remove();

                        AtomicInteger hitNumber = new AtomicInteger();
                        Map<String,Object> hitData = new HashMap<>();
                        hitData.put("startTime", timeWindow.getStart());
                        hitData.put("endTime", timeWindow.getEnd());

                        Map<String, WindowResult> hitResult = new HashMap<>();
                        comparisonResult.forEach((tagId, result) -> {
                            if(result.getHit().get()) {
                                Optional<String> tagJsonOptional = metricTagQueryService.getTagJson(tagId);
                                if(tagJsonOptional.isPresent()) {
                                    hitNumber.getAndIncrement();
                                    String tagJson = metricTagQueryService.getTagJson(tagId).get();
                                    hitResult.put(tagJson, result);
                                }
                            }
                        });
                        hitData.put("number", hitNumber.get());
                        hitData.put("result", hitResult);

                        RulePoint rulePoint = new RulePoint();
                        rulePoint.setRuleId(rule.getId());
                        rulePoint.setEventTime(DateUtil.getDateTimeOfTimestamp(timeWindow.getStart()));
                        rulePoint.setHitNumber(hitNumber.get());
                        rulePoint.setHitData(objectMapper.writeValueAsString(hitData));

                        rulePointCommandService.save(rulePoint).subscribe();
                        log.info(rulePoint.toString());
                    }
                }
            }
        } catch (Exception ex) {
            log.error("rule task error,", ex);
        }
        if (rule.getWindowType() == WindowType.MONTH
                || rule.getWindowType() == WindowType.YEAR) {
            ruleScheduleServiceImpl.removeSchedule(rule);
            ruleScheduleServiceImpl.startMonthAndYearSchedule(rule);
        }
    }

    @Getter
    @NoArgsConstructor
    class Result {

        private Map<String, Double> metricValue = new HashMap<>();

        private AtomicBoolean hit = new AtomicBoolean(false);

        @Override
        public String toString() {
            return "Result{" +
                    "metricValue=" + metricValue +
                    ", hit=" + hit +
                    '}';
        }
    }
}
